<?php
/*
	本插件：$this->model('mcp')->方法名();
	其他地方调用：model('jy_agentmcp/mcp')->方法名();
 * MCP 客户端实现（优化版）
 * 参考协议：https://modelcontextprotocol.io/specification/2024-11-05
*/
namespace jy_agentmcp;
defined('BY_JYA') or exit('error');
class mcp2 extends \plugin{
    private $mcpurl; // 存储MCP服务地址
    private $supportedProtocolVersions = ['2024-11-05', '2023-10-01']; // 支持的协议版本
    private $negotiatedProtocolVersion = ''; // 协商后的协议版本
    private $endpointId = 0; // 端点ID（用于日志）
    private $sessionId; // 会话ID
    private $isSimplifiedFormat = false; // 是否使用简化消息格式
    private $wsModel; // WebSocket实例
    private $toolListCache = []; // 工具列表缓存 [cacheKey => {tools, nextCursor}]
    private $lastActivityTime = 0; // 最后活动时间（用于心跳）

    /**
     * 处理客户绑定MCP接入点（小程序调用）
     */
    public function bindMcpEndpoint($agent, $isdemo = 0) {
        if (empty($agent['mcpurl'])) {
            return $this->formatResult(false, 'MCP地址不能为空');
        }
        $agent['mcpurl']='wss://api.xiaozhi.me/mcp/?token=eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VySWQiOjE2ODMyNywiYWdlbnRJZCI6NjQxNjI1LCJlbmRwb2ludElkIjoiYWdlbnRfNjQxNjI1IiwicHVycG9zZSI6Im1jcC1lbmRwb2ludCIsImlhdCI6MTc1ODUyODEyNywiZXhwIjoxNzkwMDg1NzI3fQ.-6qncZWHGj3K2sz_WegMT8QD2oxuSTnk8DVdWYs_pQVP1eTK3xxF3bEP5-xgQ4SgUghSeC8ClFO2W_LBs8_BAA';
        $params = [
            'mcpurl' => $agent['mcpurl'],
            'uniacid' => $_SESSION['uniacid'] ?? 0
        ];
        $isdemo && $params['mid'] = 0;
        $this->mcpurl = $agent['mcpurl'];
        
        // 检查绑定记录
        $isadd = pdo_get('jy_agentmcp_usergoods', $params);
        if (!$isadd) {
           // return $this->formatResult(false, '未找到绑定记录');
        }

        // 终止旧进程
        if ($isadd['status'] == 1 && $isadd['pid'] > 0) {
            $this->terminateProcess($isadd['pid']);
        }

		pdo_update('jy_agentmcp_usergoods',array('status'=>1),array('mcpurl'=>$agent['mcpurl'],'uniacid'=>$_SESSION['uniacid']));
		// 启动新的MCP服务进程
		$pid = $this->startMcpServerForClient($isadd['id'], $agent['mcpurl']);
		if ($pid > 0) {
			//pdo_update('jy_agentmcp_usergoods',array('pid'=>$pid,'status'=>1),array('mcpurl'=>$agent['mcpurl'],'uniacid'=>$_SESSION['uniacid']));
		} elseif($pid<0){
			pdo_update('jy_agentmcp_usergoods',array('status'=>0),array('mcpurl'=>$agent['mcpurl'],'uniacid'=>$_SESSION['uniacid']));
		}
		return '';
	}

    /**
     * 启动MCP服务进程
     */
    public function startMcpServerForClient($endpointId, $endpointUrl) {
        if (!function_exists('pcntl_fork')) {
            //$this->logError("未启用pcntl扩展，无法创建进程");
           // return -1;
        }

        $pid = pcntl_fork();
        $this->logInfo("fork结果: {$pid} (endpointId: {$endpointId})");

        if ($pid == -1) {
            $this->logError("fork失败: " . pcntl_get_last_error());
            return -1;
        } elseif ($pid == 0) {
            // 子进程：设置信号处理
            pcntl_signal(SIGTERM, function () {
                $this->logInfo("子进程 " . getmypid() . " 收到终止信号");
                exit(0);
            });
            
            $mypid = getmypid();
            $this->logInfo("子进程 {$mypid} 启动成功");
            $this->runMcpServer($endpointId, $endpointUrl, $mypid);
            exit(0);
        }
        
        return $pid;
    }

	/**
     * 运行MCP服务（子进程逻辑）
	 */
	public function runMcpServer($endpointId=0, $endpointUrl='',$mypid=0) {
        $this->endpointId = $endpointId; // 保存端点ID
		pdo_update('jy_agentmcp_usergoods',array('pid'=>$mypid,'status'=>1),array('mcpurl'=>$this->mcpurl,'uniacid'=>$_SESSION['uniacid']));
        $this->wsModel = $this->model('websocket', $endpointUrl);
        $requestId = 1;
        $reconnectDelay = 5;
        $maxReconnectDelay = 60;
        $heartbeatInterval = 20; // 心跳间隔(秒)

        while (true) {
            // 建立连接
            if (!$this->wsModel->connect()) {
                $this->logInfo("连接失败，{$reconnectDelay}秒后重试 (endpointId: {$this->endpointId})");
                sleep($reconnectDelay);
                $reconnectDelay = min($reconnectDelay * 2, $maxReconnectDelay);
                continue;
            }

            // 重置连接状态
            $reconnectDelay = 5;
            $this->lastActivityTime = time();
            $this->sessionId = uniqid('mcp_', true);
            $this->isSimplifiedFormat = false;
            $this->toolListCache = []; // 重置缓存
            $this->logInfo("已建立连接，会话ID: {$this->sessionId} (endpointId: {$this->endpointId})");
            
            // 发送hello消息
            $this->sendHelloMessage($requestId++);
            
            // 消息处理循环
            while (true) {
                try {
                    // 发送心跳
                    if (time() - $this->lastActivityTime > $heartbeatInterval) {
                        $this->sendHeartbeat($requestId++);
                        $this->lastActivityTime = time();
                    }

                    // 接收消息（非阻塞，超时5秒）
                    $response = $this->wsModel->receive(5);
                    if ($response === false) {
                        throw new \Exception("连接断开");
                    }

                    if (!empty($response)) {
                        $this->lastActivityTime = time();
                        $cleanResponse = $this->cleanInvalidCharacters($response);
                        $this->handleReceivedMessage($cleanResponse, $requestId);
                        $requestId++;
                    }

                    usleep(100000); // 100ms间隔
                } catch (\Exception $e) {
                    $this->logError("处理错误: " . $e->getMessage() . "，尝试重连");
                    $this->wsModel->close();
                    break;
                }
            }
        }
    }

    /**
     * 发送Hello消息
     */
    private function sendHelloMessage($requestId) {
        $helloMsg = $this->wrapMcpMessage([
            'jsonrpc' => '2.0',
            'method' => 'hello',
            'params' => [
                'mcp' => true,
                'supported_versions' => $this->supportedProtocolVersions
            ],
            'id' => $requestId
        ]);
        $this->wsModel->send(json_encode($helloMsg));
        $this->logInfo("发送Hello消息 (endpointId: {$this->endpointId})");
    }

    /**
     * 发送心跳包
     */
    private function sendHeartbeat($requestId) {
        $heartbeat = $this->wrapMcpMessage([
            'jsonrpc' => '2.0',
            'method' => 'heartbeat',
            'id' => $requestId
        ]);
        $this->wsModel->send(json_encode($heartbeat));
        $this->logInfo("发送心跳包 (endpointId: {$this->endpointId})");
    }

    /**
     * 包装MCP消息
     */
    private function wrapMcpMessage($payload) {
        return [
            'session_id' => $this->sessionId,
            'type' => 'mcp',
            'payload' => $payload
        ];
    }

    /**
     * 清理无效字符
     */
    private function cleanInvalidCharacters($data) {
        return preg_replace('/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/', '', $data);
    }

    /**
     * 处理接收到的消息
     */
    private function handleReceivedMessage($response, &$requestId) {
        // 编码转换
        $encoding = mb_detect_encoding($response, ['UTF-8', 'ISO-8859-1', 'Windows-1252']);
        if ($encoding && $encoding !== 'UTF-8') {
            $response = mb_convert_encoding($response, 'UTF-8', $encoding);
        }
        
        $this->logInfo("收到原始消息: {$response} (endpointId: {$this->endpointId})");
        $message = json_decode($response, true);
        
        if (json_last_error() !== JSON_ERROR_NONE) {
            $this->logError("无效的JSON消息: " . json_last_error_msg());
            return;
        }

        // 兼容简化格式消息
        if (!isset($message['type']) && isset($message['jsonrpc']) && isset($message['method'])) {
            $this->isSimplifiedFormat = true;
            $message = [
                'type' => 'mcp',
                'session_id' => $this->sessionId,
                'payload' => $message
            ];
            $this->logInfo("检测到简化格式消息，已自动转换");
        }

        // 验证消息结构
        if (!isset($message['type'], $message['payload'], $message['session_id'])) {
            $this->logError("消息结构不完整，缺少必要字段");
            return;
        }

        // 忽略非MCP消息
        if ($message['type'] !== 'mcp') {
            $this->logInfo("忽略非MCP类型消息: {$message['type']}");
            return;
        }

        $payload = $message['payload'];
        // 处理设备通知（无ID）
        if (!isset($payload['id'])) {
            $this->handleNotification($payload);
            return;
        }

        // 处理常规请求
        if (!isset($payload['method'])) {
            $this->logError("请求缺少method字段");
            $this->replyError($payload['id'], "请求缺少method字段", -32600);
            return;
        }

        $result = $this->handleMcpRequest($payload);
        if ($result !== null) {
            $this->sendResponse($result);
        }
    }

    /**
     * 发送响应消息
     */
    private function sendResponse($result) {
        if ($this->isSimplifiedFormat) {
            $this->wsModel->send(json_encode($result));
        } else {
            $responseMsg = $this->wrapMcpMessage($result);
            $this->wsModel->send(json_encode($responseMsg));
        }
    }

    /**
     * 处理设备主动通知
     */
    private function handleNotification($payload) {
        if (!isset($payload['method']) || strpos($payload['method'], 'notifications/') !== 0) {
            $this->logError("无效的通知格式: " . json_encode($payload));
            return;
        }
        
        $this->logInfo("收到设备通知: {$payload['method']}");
        // 此处可添加通知处理逻辑（如状态更新、事件记录等）
    }

    /**
     * 处理MCP请求
     */
    public function handleMcpRequest($request) {
        $response = [
            'jsonrpc' => '2.0',
            'id' => $request['id'] ?? null
        ];
        $this->logInfo("处理MCP请求 {$request['method']}");
        try {
            switch ($request['method']) {
                // 协议初始化
                case 'initialize':
                    $this->handleInitialize($request, $response);
                    break;
                    
                // 工具列表查询
                case 'tools/list':
                    $this->handleToolsList($request, $response);
                    break;
                    
                // 工具调用
                case 'tools/call':
                    $this->handleToolCall($request, $response);
                    break;
                    
                // 心跳响应（如果设备主动发送心跳请求）
                case 'heartbeat':
                    $response['result'] = ['status' => 'ok'];
                    break;
                    
                default:
                    throw new \Exception("未实现的方法: {$request['method']}", -32601);
            }
        } catch (\Exception $e) {
            $response['error'] = [
                'code' => $e->getCode() ?: -32603,
                'message' => $e->getMessage()
            ];
        }

        return $response;
    }

    /**
     * 处理initialize请求
     */
    private function handleInitialize($request, &$response) {
        $clientProtocolVersion = $request['params']['protocolVersion'] ?? '';
        $supportedVersions = $this->supportedProtocolVersions;
        
        if (empty($clientProtocolVersion)) {
            throw new \Exception("缺少协议版本参数", -32602);
        }
        
        if (!in_array($clientProtocolVersion, $supportedVersions)) {
            $suggestedVersion = end($supportedVersions);
            throw new \Exception(
                "不支持的协议版本: {$clientProtocolVersion}，支持版本: " . implode(', ', $supportedVersions) . 
                "，建议使用: {$suggestedVersion}",
                -32602
            );
        }

        $this->negotiatedProtocolVersion = $clientProtocolVersion;
        $appDesc = $this->getAppDescription(); // 获取应用描述信息
        
        $response['result'] = [
            'protocolVersion' => $clientProtocolVersion,
            'capabilities' => ['tools' => new \stdClass()],
            'serverInfo' => [
                'name' => defined('BOARD_NAME') ? BOARD_NAME : 'Unknown Device',
                'version' => $appDesc['version'] ?? '1.0.0'
            ]
        ];
    }
    /**
     * 终止进程并回收
     */
    private function terminateProcess($pid) {
        if (!$pid || !is_numeric($pid)) {
            $this->logError("无效的进程ID: {$pid}");
            return;
        }
        
        // 先检查进程是否存在
        if (!posix_getpgid($pid)) {
            $this->logInfo("进程 {$pid} 不存在，无需终止");
            return;
        }
        
        if (posix_kill($pid, SIGTERM)) {
            $this->logInfo("已发送终止信号给进程 {$pid}");
            // 等待进程退出（最多等待3秒）
            $timeout = 3;
            $start = time();
            do {
                pcntl_waitpid($pid, $status, WNOHANG);
                if (!posix_getpgid($pid)) { // 检查进程是否存在
                    break;
                }
                usleep(100000);
            } while (time() - $start < $timeout);
            
            if (posix_getpgid($pid)) {
                $this->logError("进程 {$pid} 未正常退出，强制终止");
                posix_kill($pid, SIGKILL);
            }
        } else {
            $this->logError("无法终止进程 {$pid}，可能已退出");
        }
    }
    /**
     * 处理工具列表查询
     */
    private function handleToolsList($request, &$response) {
        $cursor = $request['params']['cursor'] ?? '';
        $listUserOnly = isset($request['params']['withUserTools']) ? (bool)$request['params']['withUserTools'] : false;
        $cacheKey = md5($cursor . ($listUserOnly ? 1 : 0));

        // 检查缓存
        if (isset($this->toolListCache[$cacheKey])) {
            $response['result'] = $this->toolListCache[$cacheKey];
            return;
        }

        // 实际项目中应从设备或数据库获取工具列表
        $tools = $this->getToolsFromDevice($cursor, $listUserOnly, $nextCursor);
        
        $result = ['tools' => $tools];
        if (!empty($nextCursor)) {
            $result['nextCursor'] = $nextCursor;
        }

        // 缓存结果（有效期5分钟）
        $this->toolListCache[$cacheKey] = $result;
        $response['result'] = $result;
    }

    /**
     * 处理工具调用
     */
    private function handleToolCall($request, &$response) {
        $params = $request['params'] ?? [];
        if (empty($params['name'])) {
            throw new \Exception("缺少工具名称", -32602);
        }

        $toolName = $params['name'];
        $arguments = $params['arguments'] ?? [];

        // 验证工具是否存在
        if (!$this->isToolExists($toolName)) {
            throw new \Exception("工具不存在: {$toolName}", -32601);
        }

        // 验证参数
        $this->validateToolParams($toolName, $arguments);

        // 执行工具调用
        $result = $this->executeTool($toolName, $arguments);
        
        $response['result'] = [
            'content' => [['type' => 'text', 'text' => json_encode($result)]],
            'isError' => false
        ];
    }

    /**
     * 回复错误消息
     */
    private function replyError($id, $message, $code = -32603) {
        $payload = [
            'jsonrpc' => '2.0',
            'id' => $id,
            'error' => [
                'code' => $code,
                'message' => $message
            ]
        ];
        
        $this->sendResponse($payload);
    }

    /**
     * 格式化通用返回结果
     */
    private function formatResult($success, $message, $data = []) {
        return [
            'success' => $success,
            'message' => $message,
            'data' => $data
        ];
    }
    /**
     * 获取应用描述（可根据实际项目实现）
     */
    private function getAppDescription() {
        return [
            'version' => '1.0.0',
            'build' => '20240601'
        ];
    }

    /**
     * 从设备获取工具列表（实际实现需对接设备）
     */
    private function getToolsFromDevice($cursor, $listUserOnly, &$nextCursor) {
        // 示例实现，实际应从设备或数据库查询
        $allTools = [
            [
                'name' => 'get_latest_music',
                'description' => '获取最新音乐',
                'inputSchema' => [
                    'properties' => [
                        'limit' => [
                            'type' => 'integer',
                            'description' => '获取数量'
                        ],
                        'page' => [
                             'type' => 'integer',
                            'description' => '页码'
                        ]
                    ],
                    'required' => ['limit','page']
                ]
            ],
            [
                'name' => 'search_music',
                'description' => '搜索音乐',
                'inputSchema' => [
                    'properties' => [
                        'query' => [
                            'type' => 'string',
                            'description' => '搜索关键词'
                        ],
                        'page' => [
                            'type' => 'integer',
                            'description' => '页码'
                        ]
                    ],
                    'required' => ['query','page']
                ]
            ],
             [
                'name' => 'play_music',
                'description' => '播放指定音乐',
                'inputSchema' => [
                    'properties' => [
                        'music_id' => [
                            'type' => 'integer',
                            'description' => '音乐ID'
                        ]
                    ],
                    'required' => ['music_id']
                ]
            ]
        ];
        // 分页逻辑（示例）
        $pageSize = 10;
        $startIndex = $cursor ? array_search($cursor, array_column($allTools, 'name')) + 1 : 0;
        $tools = array_slice($allTools, $startIndex, $pageSize);
        $nextCursor = (count($allTools) > $startIndex + $pageSize) ? $allTools[$startIndex + $pageSize]['name'] : '';
        return $tools;
    }

    /**
     * 检查工具是否存在（实际实现需对接设备）
     */
    private function isToolExists($toolName) {
        // 示例实现
        $existsTools = ['self.get_device_status', 'self.audio_speaker.set_volume'];
        return in_array($toolName, $existsTools);
    }

    /**
     * 验证工具参数（实际实现需对接设备）
     */
    private function validateToolParams($toolName, $params) {
        // 示例：验证音量设置参数
        if ($toolName === 'self.audio_speaker.set_volume') {
            if (!isset($params['volume']) || !is_numeric($params['volume'])) {
                throw new \Exception("无效的音量参数", -32602);
            }
            $volume = (int)$params['volume'];
            if ($volume < 0 || $volume > 100) {
                throw new \Exception("音量必须在0-100之间", -32602);
            }
        }
    }

    /**
     * 执行工具调用（实际实现需对接设备）
     */
    private function executeTool($toolName, $arguments) {
        // 示例实现
        switch ($toolName) {
            case 'get_latest_music':
                $limit = isset($arguments['limit']) ? intval($arguments['limit']) : 5;
				$params=array('uniacid'=>$_SESSION['uniacid']);
				$pindex = max(1, intval($arguments['page'] ?? 1));
				$psize = $limit;
				$total='';
				$list=pdo_getslice('jy_agentmcp_mp3',$params,array($pindex,$psize),$total,$fields,'','id desc');
				if($list){
					foreach($list as &$item){
					}
				}
				return [
					'success' => true,
					'music_list' => $list
				];
            case 'search_music':
                if (!isset($arguments['query']) || empty($arguments['query'])) {
					throw new \Exception("搜索关键词不能为空");
				}
				$query = "%" . $arguments['query'] . "%";
				$params=array('uniacid'=>$_SESSION['uniacid']);
				if(!empty($arguments['title'])){
					//$params['title like']=$query;
				}
				$pindex = max(1, intval($arguments['page'] ?? 1));
				$psize = 10;
				$total='';
				$list=pdo_getslice('jy_agentmcp_mp3',$params,array($pindex,$psize),$total,$fields,'','id desc');
				if($list){
					foreach($list as &$item){

					}
				}
				return [
					'success' => true,
					'count' => count($list),
					'music_list' => $list
				];
            case 'play_music':
               if (empty($arguments['music_id'])) {
					throw new \Exception("音乐ID不能为空");
				}
				$musicId = intval($arguments['music_id']);
				$music = pdo_get('jy_agentmcp_mp3',array('id'=>$musicId,'uniacid'=>$_SESSION['uniacid']));
                $music['url']=tomedia($music['url']);
                $music=array(
                    'id'=>1,
                    'title'=>'凡人歌',
                    'artist'=>'网络',
                    'url'=>tomedia('oss/1/audios/2025/09/mcNfQhKIyd.mp3'),
                    'album'=>'凡人歌',
                    'genre'=>'抒情',
                );
				if (!$music['id']) {
					throw new \Exception("未找到ID为{$musicId}的音乐");
				}
				return[
					'success' => true,
					'message' => "正在播放: {$music['title']} - {$music['artist']}",
					'music' => $music
				];
            default:
                throw new \Exception("工具执行逻辑未实现", -32603);
        }
    }
    /**
     * 日志记录（信息）
     */
    private function logInfo($message) {
        $this->log("[INFO] [{$this->endpointId}] " . $message);
    }

    /**
     * 日志记录（错误）
     */
    private function logError($message) {
        $this->log("[ERROR] [{$this->endpointId}] " . $message);
    }

    /**
     * 通用日志方法
     */
    private function log($message) {
        $logFile = ROOT_D . '/addons/jy_agentmcp/mcp_' . date('Ymd') . '.log';
        $dir = dirname($logFile);
        if (!is_dir($dir)) {
            mkdir($dir, 0755, true);
        }
        $logContent = date('Y-m-d H:i:s') . " {$message}\n";
        file_put_contents($logFile, $logContent, FILE_APPEND);
    }
}
?>